package org.example;
import com.typesafe.config.ConfigFactory;
import org.apache.pekko.actor.*;
import org.apache.pekko.pattern.Patterns;
import org.apache.pekko.util.Timeout;
import scala.concurrent.ExecutionContext;
import scala.concurrent.Future;
import org.apache.pekko.dispatch.OnComplete;

import java.io.*;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;

public class Master extends AbstractActor {
    private ExecuteExpression executeExpression ;
    public List<ActorSelection> workers = new CopyOnWriteArrayList<>();
    public static void main(String[] args) {
        ActorSystem actorSystem = ActorSystem.create("actorSystem", ConfigFactory.load());
        actorSystem.actorOf(Props.create(Master.class) , "master");
    }
    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(Message.class , this::messageHandler)
                .matchAny(this::handler)
                .build();
    }

    private void messageHandler(Message message) {
        switch (message.getType()) {
            case REGISTER:
                System.out.println("register");
                String url = (String)message.getMessage();
                System.out.println(url);
                // 获取发送端的引用
                ActorRef self = getSender();
                System.out.println("address: " + self.path());

                ActorSelection actorSelection = context().actorSelection(url);
                actorSelection.tell("handshake" , getSelf());

                workers.add(actorSelection);
                String script = "int a = 0 ; " +
                                "int b = 9 ; " +
                                "int c = a > b ? a : b;" +
                        "System.out.println(\"max: \" + c);" ;

                ExecuteExpression.Builder builder = new ExecuteExpression.Builder();
                ExecuteExpression ee = builder.builder(actorSelection);
                System.out.println("这里就是远程调用");
                ee.execute(script , new HashMap<String,Object>(){
                    {
                        put("aaa" , "");
                    }
                });
                break;
            default:
                break;
        }
    }
    public void handler(Object o){
        if(o instanceof String){
            String msg = o.toString();
            // 以 register 开头，是注册新的 work
            if(msg.startsWith("register")){
                String[] split = msg.split("::");
                System.out.println(split[1]);
                ActorSelection actorSelection = context().actorSelection(split[1]);
                workers.add(actorSelection);
                actorSelection.tell("handshake" , getSelf());
                for (ActorSelection worker : workers) {
                    System.out.println("worker: " + worker.pathString());
                }
            }else if(msg.startsWith("start")){
                // 客户端发来请求，开始计算任务
                // 需要给出的是 源文件的地址
                String[] path = msg.split("::");
                try {
                    FileReader reader = new FileReader(path[1]);
                    BufferedReader bufferedReader = new BufferedReader(reader);

                    String line = null;
                    while ((line = bufferedReader.readLine()) != null) {
                        String[] split = line.split("\\s+");
                        for (String word : split) {
                            int index = word.hashCode()%workers.size();
                            workers.get(index).tell(word , getSelf());
                        }
                    }
                    for (ActorSelection worker : workers) {
                        worker.tell("print" , getSelf());
                    }

                } catch (FileNotFoundException e) {
                    e.printStackTrace();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}
